#include <boost/asio.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <sys/eventfd.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <unistd.h>
#include <iostream>
#include <thread>
#include <vector>
#include <string>
#include <cstring>
#include <atomic>
#include <chrono>

using namespace boost::asio;

/**
 * 基于共享内存的EventFD注册中心
 * 
 * 功能特性：
 * 1. 支持多topic的发布-订阅模式
 * 2. 订阅者进程注册时填写eventfd
 * 3. 发布者进程定期更新，通知所有订阅者
 * 4. 支持动态添加/移除订阅者
 * 5. 进程崩溃自动清理机制
 */

constexpr size_t MAX_TOPICS = 64;           // 最大topic数量
constexpr size_t MAX_SUBSCRIBERS = 32;      // 每个topic最大订阅者数量
constexpr size_t TOPIC_NAME_SIZE = 64;      // topic名称最大长度
constexpr size_t PROCESS_NAME_SIZE = 32;    // 进程名称最大长度

// 订阅者信息
struct SubscriberInfo {
    pid_t pid;                              // 进程ID
    int eventfd;                            // EventFD句柄
    char process_name[PROCESS_NAME_SIZE];   // 进程名称
    std::atomic<uint64_t> last_heartbeat;  // 最后心跳时间戳
    std::atomic<bool> active;              // 是否活跃
    
    SubscriberInfo() {
        reset();
    }
    
    void reset() {
        pid = 0;
        eventfd = -1;
        memset(process_name, 0, PROCESS_NAME_SIZE);
        last_heartbeat.store(0);
        active.store(false);
    }
    
    bool is_valid() const {
        return active.load() && pid > 0 && eventfd >= 0;
    }
    
    void update_heartbeat() {
        auto now = std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        last_heartbeat.store(now);
    }
    
    bool is_alive(uint64_t timeout_seconds = 30) const {
        if (!active.load()) return false;
        
        auto now = std::chrono::duration_cast<std::chrono::seconds>(
            std::chrono::system_clock::now().time_since_epoch()).count();
        return (now - last_heartbeat.load()) < timeout_seconds;
    }
};

// Topic信息
struct TopicInfo {
    char topic_name[TOPIC_NAME_SIZE];       // Topic名称
    std::atomic<bool> active;              // 是否活跃
    std::atomic<int> subscriber_count;     // 订阅者数量
    SubscriberInfo subscribers[MAX_SUBSCRIBERS]; // 订阅者列表
    std::atomic<uint64_t> message_count;   // 消息计数器
    
    TopicInfo() {
        reset();
    }
    
    void reset() {
        memset(topic_name, 0, TOPIC_NAME_SIZE);
        active.store(false);
        subscriber_count.store(0);
        message_count.store(0);
        for (auto& sub : subscribers) {
            sub.reset();
        }
    }
    
    bool is_valid() const {
        return active.load() && strlen(topic_name) > 0;
    }
};

// 共享内存注册表结构
struct EventFdRegistry {
    std::atomic<bool> initialized;         // 是否已初始化
    std::atomic<int> active_topics;        // 活跃topic数量
    std::atomic<uint64_t> global_version;  // 全局版本号
    TopicInfo topics[MAX_TOPICS];          // Topic列表
    
    EventFdRegistry() {
        initialized.store(false);
        active_topics.store(0);
        global_version.store(0);
        for (auto& topic : topics) {
            topic.reset();
        }
    }
    
    void initialize() {
        if (!initialized.exchange(true)) {
            std::cout << "EventFD注册表初始化完成" << std::endl;
        }
    }
};

/**
 * EventFD注册中心管理器
 */
class EventFdRegistryManager {
private:
    static constexpr const char* SHM_NAME = "/eventfd_registry_shm";
    int shm_fd_;
    EventFdRegistry* registry_;
    bool is_owner_;
    
public:
    EventFdRegistryManager() : shm_fd_(-1), registry_(nullptr), is_owner_(false) {
    }
    
    ~EventFdRegistryManager() {
        cleanup();
    }
    
    bool initialize() {
        // 尝试创建共享内存
        shm_fd_ = shm_open(SHM_NAME, O_CREAT | O_EXCL | O_RDWR, 0666);
        if (shm_fd_ != -1) {
            is_owner_ = true;
            std::cout << "创建新的共享内存注册表" << std::endl;
        } else {
            // 打开现有共享内存
            shm_fd_ = shm_open(SHM_NAME, O_RDWR, 0666);
            if (shm_fd_ == -1) {
                perror("shm_open");
                return false;
            }
            std::cout << "连接到现有共享内存注册表" << std::endl;
        }
        
        if (ftruncate(shm_fd_, sizeof(EventFdRegistry)) == -1) {
            perror("ftruncate");
            return false;
        }
        
        registry_ = static_cast<EventFdRegistry*>(mmap(
            nullptr, 
            sizeof(EventFdRegistry),
            PROT_READ | PROT_WRITE,
            MAP_SHARED,
            shm_fd_, 0
        ));
        
        if (registry_ == MAP_FAILED) {
            perror("mmap");
            return false;
        }
        
        if (is_owner_) {
            // 初始化注册表
            new (registry_) EventFdRegistry();
        }
        
        registry_->initialize();
        return true;
    }
    
    void cleanup() {
        if (registry_ && registry_ != MAP_FAILED) {
            munmap(registry_, sizeof(EventFdRegistry));
            registry_ = nullptr;
        }
        
        if (shm_fd_ != -1) {
            close(shm_fd_);
            shm_fd_ = -1;
        }
        
        if (is_owner_) {
            shm_unlink(SHM_NAME);
        }
    }
    
    EventFdRegistry* get_registry() const {
        return registry_;
    }
};

/**
 * 订阅者类
 */
class EventFdSubscriber {
private:
    EventFdRegistryManager registry_manager_;
    EventFdRegistry* registry_;
    io_context& io_;
    std::vector<std::unique_ptr<posix::stream_descriptor>> descriptors_;
    std::vector<std::string> subscribed_topics_;
    std::string process_name_;
    std::atomic<bool> running_;
    std::thread heartbeat_thread_;
    
public:
    EventFdSubscriber(io_context& io, const std::string& process_name) 
        : registry_(nullptr), io_(io), process_name_(process_name), running_(true) {
        
        if (!registry_manager_.initialize()) {
            throw std::runtime_error("Failed to initialize registry manager");
        }
        
        registry_ = registry_manager_.get_registry();
        start_heartbeat();
    }
    
    ~EventFdSubscriber() {
        running_ = false;
        if (heartbeat_thread_.joinable()) {
            heartbeat_thread_.join();
        }
        
        // 清理订阅
        for (const auto& topic : subscribed_topics_) {
            unsubscribe(topic);
        }
    }
    
    bool subscribe(const std::string& topic_name) {
        if (topic_name.length() >= TOPIC_NAME_SIZE) {
            std::cerr << "Topic名称过长: " << topic_name << std::endl;
            return false;
        }
        
        // 创建EventFD
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd == -1) {
            perror("eventfd");
            return false;
        }
        
        // 查找或创建topic
        TopicInfo* topic = find_or_create_topic(topic_name);
        if (!topic) {
            close(efd);
            return false;
        }
        
        // 注册订阅者
        bool registered = false;
        for (auto& subscriber : topic->subscribers) {
            if (!subscriber.active.load()) {
                subscriber.pid = getpid();
                subscriber.eventfd = efd;
                strncpy(subscriber.process_name, process_name_.c_str(), PROCESS_NAME_SIZE - 1);
                subscriber.update_heartbeat();
                subscriber.active.store(true);
                
                topic->subscriber_count.fetch_add(1);
                registered = true;
                break;
            }
        }
        
        if (!registered) {
            std::cerr << "Topic订阅者已满: " << topic_name << std::endl;
            close(efd);
            return false;
        }
        
        // 创建异步描述符
        auto descriptor = std::make_unique<posix::stream_descriptor>(io_, efd);
        descriptors_.push_back(std::move(descriptor));
        subscribed_topics_.push_back(topic_name);
        
        std::cout << "成功订阅topic: " << topic_name << " (EventFD: " << efd << ")" << std::endl;
        
        // 开始异步等待事件
        start_async_wait(topic_name, descriptors_.back().get());
        
        return true;
    }
    
    void unsubscribe(const std::string& topic_name) {
        TopicInfo* topic = find_topic(topic_name);
        if (!topic) return;
        
        pid_t my_pid = getpid();
        for (auto& subscriber : topic->subscribers) {
            if (subscriber.active.load() && subscriber.pid == my_pid) {
                close(subscriber.eventfd);
                subscriber.reset();
                topic->subscriber_count.fetch_sub(1);
                std::cout << "取消订阅topic: " << topic_name << std::endl;
                break;
            }
        }
    }
    
private:
    TopicInfo* find_topic(const std::string& topic_name) {
        for (auto& topic : registry_->topics) {
            if (topic.is_valid() && strcmp(topic.topic_name, topic_name.c_str()) == 0) {
                return &topic;
            }
        }
        return nullptr;
    }
    
    TopicInfo* find_or_create_topic(const std::string& topic_name) {
        // 先查找现有topic
        TopicInfo* existing = find_topic(topic_name);
        if (existing) return existing;
        
        // 创建新topic
        for (auto& topic : registry_->topics) {
            if (!topic.active.load()) {
                strncpy(topic.topic_name, topic_name.c_str(), TOPIC_NAME_SIZE - 1);
                topic.active.store(true);
                registry_->active_topics.fetch_add(1);
                std::cout << "创建新topic: " << topic_name << std::endl;
                return &topic;
            }
        }
        
        std::cerr << "无法创建新topic，已达到最大数量" << std::endl;
        return nullptr;
    }
    
    void start_async_wait(const std::string& topic_name, posix::stream_descriptor* descriptor) {
        auto buffer = std::make_shared<uint64_t>();
        
        descriptor->async_read_some(
            boost::asio::buffer(buffer.get(), sizeof(*buffer)),
            [this, topic_name, descriptor, buffer](const boost::system::error_code& ec, std::size_t bytes) {
                if (!running_) return;
                
                if (!ec && bytes == sizeof(*buffer)) {
                    std::cout << "收到topic [" << topic_name << "] 事件, 值: " << *buffer << std::endl;
                    
                    // 继续等待下一个事件
                    start_async_wait(topic_name, descriptor);
                } else if (ec != boost::asio::error::operation_aborted) {
                    std::cerr << "EventFD读取错误: " << ec.message() << std::endl;
                    
                    // 错误重试
                    std::this_thread::sleep_for(std::chrono::milliseconds(100));
                    start_async_wait(topic_name, descriptor);
                }
            }
        );
    }
    
    void start_heartbeat() {
        heartbeat_thread_ = std::thread([this]() {
            while (running_) {
                update_heartbeat();
                std::this_thread::sleep_for(std::chrono::seconds(10));
            }
        });
    }
    
    void update_heartbeat() {
        pid_t my_pid = getpid();
        
        for (auto& topic : registry_->topics) {
            if (!topic.is_valid()) continue;
            
            for (auto& subscriber : topic.subscribers) {
                if (subscriber.active.load() && subscriber.pid == my_pid) {
                    subscriber.update_heartbeat();
                }
            }
        }
    }
};

/**
 * 发布者类
 */
class EventFdPublisher {
private:
    EventFdRegistryManager registry_manager_;
    EventFdRegistry* registry_;
    std::string process_name_;
    
public:
    EventFdPublisher(const std::string& process_name) 
        : registry_(nullptr), process_name_(process_name) {
        
        if (!registry_manager_.initialize()) {
            throw std::runtime_error("Failed to initialize registry manager");
        }
        
        registry_ = registry_manager_.get_registry();
    }
    
    bool publish(const std::string& topic_name, uint64_t value = 1) {
        TopicInfo* topic = find_topic(topic_name);
        if (!topic) {
            std::cerr << "Topic不存在: " << topic_name << std::endl;
            return false;
        }
        
        int notified_count = 0;
        
        // 通知所有活跃订阅者
        for (auto& subscriber : topic->subscribers) {
            if (!subscriber.is_valid() || !subscriber.is_alive()) {
                continue;
            }
            
            ssize_t result = write(subscriber.eventfd, &value, sizeof(value));
            if (result == sizeof(value)) {
                notified_count++;
            } else {
                std::cerr << "通知订阅者失败 PID:" << subscriber.pid 
                         << " EventFD:" << subscriber.eventfd << std::endl;
                
                // 标记为非活跃
                subscriber.active.store(false);
                topic->subscriber_count.fetch_sub(1);
            }
        }
        
        topic->message_count.fetch_add(1);
        
        std::cout << "发布消息到topic [" << topic_name << "], 值: " << value 
                  << ", 通知了 " << notified_count << " 个订阅者" << std::endl;
        
        return notified_count > 0;
    }
    
    void cleanup_dead_subscribers() {
        for (auto& topic : registry_->topics) {
            if (!topic.is_valid()) continue;
            
            for (auto& subscriber : topic.subscribers) {
                if (subscriber.active.load() && !subscriber.is_alive()) {
                    std::cout << "清理死亡订阅者 PID:" << subscriber.pid 
                             << " from topic:" << topic.topic_name << std::endl;
                    
                    subscriber.reset();
                    topic.subscriber_count.fetch_sub(1);
                }
            }
        }
    }
    
    void print_registry_status() {
        std::cout << "\n=== EventFD注册表状态 ===" << std::endl;
        std::cout << "活跃Topics: " << registry_->active_topics.load() << std::endl;
        std::cout << "全局版本: " << registry_->global_version.load() << std::endl;
        
        for (const auto& topic : registry_->topics) {
            if (!topic.is_valid()) continue;
            
            std::cout << "\nTopic: " << topic.topic_name << std::endl;
            std::cout << "  订阅者数量: " << topic.subscriber_count.load() << std::endl;
            std::cout << "  消息计数: " << topic.message_count.load() << std::endl;
            
            for (const auto& subscriber : topic.subscribers) {
                if (subscriber.is_valid()) {
                    std::cout << "    订阅者 - PID:" << subscriber.pid 
                             << " 进程:" << subscriber.process_name
                             << " EventFD:" << subscriber.eventfd
                             << " 活跃:" << (subscriber.is_alive() ? "是" : "否") << std::endl;
                }
            }
        }
        std::cout << "========================\n" << std::endl;
    }
    
private:
    TopicInfo* find_topic(const std::string& topic_name) {
        for (auto& topic : registry_->topics) {
            if (topic.is_valid() && strcmp(topic.topic_name, topic_name.c_str()) == 0) {
                return &topic;
            }
        }
        return nullptr;
    }
};

// 演示程序
void run_subscriber_demo() {
    std::cout << "启动订阅者进程 PID:" << getpid() << std::endl;
    
    io_context io;
    EventFdSubscriber subscriber(io, "订阅者进程");
    
    // 订阅多个topic
    subscriber.subscribe("sensor_data");
    subscriber.subscribe("system_events");
    subscriber.subscribe("user_actions");
    
    std::cout << "订阅者开始等待事件..." << std::endl;
    
    // 运行IO循环
    io.run();
}

void run_publisher_demo() {
    std::cout << "启动发布者进程 PID:" << getpid() << std::endl;
    
    EventFdPublisher publisher("发布者进程");
    
    // 等待订阅者准备就绪
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    for (int i = 1; i <= 10; ++i) {
        // 发布不同topic的消息
        publisher.publish("sensor_data", i * 10);
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        
        publisher.publish("system_events", i);
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
        
        if (i % 3 == 0) {
            publisher.publish("user_actions", i * 100);
        }
        
        // 每5次发布后清理死亡订阅者并打印状态
        if (i % 5 == 0) {
            publisher.cleanup_dead_subscribers();
            publisher.print_registry_status();
        }
        
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    
    std::cout << "发布者任务完成" << std::endl;
}

int main(int argc, char* argv[]) {
    try {
        if (argc < 2) {
            std::cout << "使用方法: " << argv[0] << " [subscriber|publisher]" << std::endl;
            return 1;
        }
        
        std::string mode = argv[1];
        
        if (mode == "subscriber") {
            run_subscriber_demo();
        } else if (mode == "publisher") {
            run_publisher_demo();
        } else {
            std::cout << "无效模式，请使用 'subscriber' 或 'publisher'" << std::endl;
            return 1;
        }
        
    } catch (const std::exception& e) {
        std::cerr << "错误: " << e.what() << std::endl;
        return 1;
    }
    
    return 0;
}


